package com.x.rocketmq;

import com.alibaba.fastjson2.JSON;
import com.x.gzip.GZIPUtils;
import com.x.rocketmq.constant.RocketMQConstants;
import com.x.rocketmq.init.RocketMQProducerInit;
import com.x.secret.AESGeneralUtil;
import com.x.secret.MD5Util;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author xjk
 * @date 2025-07-03
 */
@Component
@Slf4j
public class RocketMQProducerServer {



  private Map<String, String> ipMap = new HashMap<String, String>();


  /**
   * 秘钥key
   */
  private Map<String, String> keyMap = new HashMap<String, String>();


  /**
   * 支持客户端自行通过注入的方式发送消息
   * 解决客户端不走服务治理时，无法发送消息
   */
  @Autowired(required = false)
  private RocketMQProducerInit producer;


  public Map<String,String> sendDataMsgConcurrently(String topic, String tag, RocketMqMessageCommand command) {
    Map<String,String> resultMap = new HashMap<>();
    String newTopic = "";
    try{
      Map<String, Object> dataRowMap = validation(topic, tag, command, null);
      Message msg = (Message) dataRowMap.get("msg");
      SendResult sendResult = null;
      newTopic = msg.getTopic();
      try {
        sendResult = producer.getDefaultMQProducer().send(msg);
        if (sendResult == null || sendResult.getSendStatus() != SendStatus.SEND_OK) {
          log.warn(newTopic + command.getMsgId() + "生产者发送消息失败!" + sendResult.getSendStatus());
          resultMap.put("code", RocketMQConstants.ROCKETMQ_SEND_ERRORCODE);
          resultMap.put("msg", RocketMQConstants.ROCKETMQ_SEND_ERRORMSG + sendResult.getSendStatus());
        } else {
          resultMap.put("code", RocketMQConstants.SUCCESSCODE);
          resultMap.put("msg", RocketMQConstants.SUCCESSMSG + sendResult.getSendStatus());
        }
      } catch (Exception e) {
        resultMap.put("code", RocketMQConstants.ROCKETMQ_SEND_ERRORCODE);
        resultMap.put("msg", RocketMQConstants.ROCKETMQ_SEND_ERRORMSG + "+++status++【明细】" + e.getMessage());
        log.error("消息队列topic:" + newTopic + "消息key:" + command.getMsgId() + e.getMessage(), e);
        throw new RuntimeException(topic + "消息正向无序发送异常", e);
      }
    } catch (Exception e){
      log.error("消息发送失败，topic={} - {}",topic ,e,e);
      throw new RuntimeException("消息发送失败" + topic);
    }
    return resultMap;
  }


  /**
   * 不论是 顺序发送 还是 无序发送，都对消息数据进行校验
   */
  public Map<String, Object> validation(String topic, String tags, RocketMqMessageCommand command, String type){

    Map<String,Object> resultMap = new HashMap<>();
    String msgId = command.getMsgId();
    if (StringUtils.isEmpty(msgId)){
      throw new RuntimeException(topic+"缺少msgId参数！");
    }
    if (StringUtils.isEmpty(command.getMsgType())){
      throw new RuntimeException(topic+"缺少msgType参数！");
    }
    if (StringUtils.isEmpty(tags)){
      tags = RocketMQConstants.SUB_ALL;
    }
    command.setTopic(topic);
    command.setTag(tags);
    command.setSendTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));


    // todo: 消息数据加密
    String msgBody = JSON.toJSONString(command);
    byte[] msgByte = null;
    try
    {
      if (RocketMQConstants.IS_COMPRESSION_ZIP)
      {
        msgByte = Base64.encodeBase64(GZIPUtils.compress(messageEncryption(topic, msgBody)));
      }
      else
      {
        msgByte = messageEncryption(topic, msgBody).getBytes();
      }
    }
    catch (Exception e)
    {
      log.error(topic + " zip happen exception,Do not execute message sending business logic" + " " + msgBody, e);
      throw new RuntimeException(e);
    }

    Message msg = new Message(topic, tags, msgId, msgByte);
    resultMap.put("msg",msg);
    return null;
  }


  public String messageEncryption(String msgType, String dataMsg)
  {
    try
    {
      if (keyMap.containsKey(msgType))
      {
        String key = keyMap.get(msgType);
        String encryptStr = AESGeneralUtil.encrypt(dataMsg, key);
        return encryptStr;
      }
      else
      {
        MD5Util md5Util = new MD5Util();
        String key = md5Util.getMd5_16New(msgType.trim());
        keyMap.put(msgType, key);
        String encryptStr = AESGeneralUtil.encrypt(dataMsg, key);
        return encryptStr;
      }
    }
    catch (Exception e)
    {
      log.error("生产者消息数据包加密异常:" + e, e);
      throw new RuntimeException("生产者消息数据包加密异常" + e.getMessage());
    }
  }

}
